package com.atguigu.gmall.realtime.dwd.app;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.util.*;

public class Dimapp01 {
    public static void main(String[] args) throws Exception {
        //TODO 1.环境准备
        //1.1 获取流环境
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT, 8888);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //2.1 设置检查点存储位置
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        //2.2 设置检查点超时时间
        checkpointConfig.setCheckpointTimeout(60000L);
        //2.3 设置检查点最小间隔时间
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
        //2.4 设置job取消后检查点是否保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.5 重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        //TODO 3.从kafka的主题中读取数据
        String groupId = "dim_app_group";
        //3.1 获取kafkaSource
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setTopics(Constant.TOPIC_DB)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                //注意：如果使用的是SimpleStringSchema对kafka消息做反序列化，如果读取到的消息为空，会报错
                //.setValueOnlyDeserializer(new SimpleStringSchema())
                //如果要处理空消息的话，需要自定义反序列化器
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] massage) throws IOException {
                        if (massage != null) {
                            return new String(massage);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();
        //3.2 将kafkaSource封装为流
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_Source").setParallelism(1);
        //TODO 4.对流中数据进行类型转换并进行简单的ETL    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                        String db = jsonObject.getString("database");
                        String type = jsonObject.getString("type");
                        String data = jsonObject.getString("data");
                        if ("gmall0221".equals(db)
                                && ("insert".equals(type)
                                || "update".equals(type)
                                || "delete".equals(type)
                                || "bootstrap-insert".equals(type))
                                && data != null
                                && data.length() > 2) {
                            collector.collect(jsonObject);
                        }
                    }
                }
        );
        //jsonObjDS.print();
        //TODO 5.使用FlinkCDC从配置表中读取配置信息
        //获取mysqlSource
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList("gmall0221_config")
                .tableList("gmall0221_config.table_process_dim")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .jdbcProperties(props)
                .build();
        //将mysqlSource转换为流
        DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");
        //mysqlDS.print();
        //TODO 6.对flinkCDC读取到的数据进行类型转换  jsonStr->实体类对象
        SingleOutputStreamOperator<TableProcessDim> tpDS = mysqlDS.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String jsonStr) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                        String op = jsonObject.getString("op");
                        TableProcessDim tableProcessDim = null;
                        if ("d".equals(op)) {
                            tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                        } else {
                            tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                        }
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }
        );
        //TODO 7.在广播配置信息前，根据配置流中的配置在HBase中执行建表或者删除表操作
        tpDS = tpDS.map(
                new RichMapFunction<TableProcessDim, TableProcessDim>() {
                    Connection hBaseConnection = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public TableProcessDim map(TableProcessDim tableProcessDim) throws Exception {
                        String op = tableProcessDim.getOp();
                        String sinkTable = tableProcessDim.getSinkTable();
                        String[] families = tableProcessDim.getSinkFamily().split(",");
                        if ("c".equals(op) || "r".equals(op)) {
                            HBaseUtil.createHBaseTable(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable, families);
                        } else if ("d".equals(op)) {
                            HBaseUtil.dropHBaseTable(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable);
                        } else {
                            HBaseUtil.dropHBaseTable(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable);
                            HBaseUtil.createHBaseTable(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable, families);
                        }
                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
//        //TODO 8.广播配置流---broadcast
//        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor = new MapStateDescriptor<String, TableProcessDim>("mapStateDescriptor", String.class, TableProcessDim.class);
//        BroadcastStream<TableProcessDim> broadcastDS = tpDS.broadcast(mapStateDescriptor);
//        //TODO 9.关联主流业务数据以及广播流配置信息---connect
//        BroadcastConnectedStream<JSONObject, TableProcessDim> connectDS = jsonObjDS.connect(broadcastDS);
//        //TODO 10.对关联后的数据进行处理---process
//        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimDS = connectDS.process(
//                new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {
//                    @Override
//                    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
//                        String table = jsonObject.getString("table");
//                        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
//                        TableProcessDim tableProcessDim = broadcastState.get(table);
//                        if (tableProcessDim != null) {
//                            JSONObject dataJsonObj = JSONObject.parseObject("data");
//                            String sinkColumns = tableProcessDim.getSinkColumns();
//                            deleteNotNeedColumn(dataJsonObj, sinkColumns);
//                            String type = jsonObject.getString("type");
//                            dataJsonObj.put("type", type);
//                            collector.collect(Tuple2.of(dataJsonObj, tableProcessDim));
//                        }
//                    }
//
//                    @Override
//                    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
//                        String op = tableProcessDim.getOp();
//                        String sourceTable = tableProcessDim.getSourceTable();
//                        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);
//                        if ("d".equals(op)) {
//                            broadcastState.remove(sourceTable);
//                        } else {
//                            broadcastState.put(sourceTable, tableProcessDim);
//                        }
//                    }
//                }
//        );
//        //TODO 11.将维度数据同步到HBase对应的表中
        env.execute();
    }
//
//    private static void deleteNotNeedColumn(JSONObject dataJsonObj, String sinkColumns) {
//        String[] columnArr = sinkColumns.split(",");
//        List<String> columnList = Arrays.asList(columnArr);
//        Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();
//        Iterator<Map.Entry<String, Object>> it = entries.iterator();
//        for(;it.hasNext();){
//            Map.Entry<String, Object> entry = it.next();
//            if (!columnList.contains(entry.getKey())){
//            it.remove();
//            }
//        }
//        entries.removeIf(entry -> !columnList.contains(entry.getKey()));
//    }
}
